package com.fengye.hbase.example;

import com.sun.istack.NotNull;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.*;

/**
 * ```
 * put 'travel',Date.new().getTime(),'f:content','{"p_id":"100001","p_name":"<张家界-凤凰-天门山-玻璃栈道5日游>","price":"2141"}'
 * ```
 * 这样一行数据,将 content 中 f_id 的值解析为 rowKey,content做为列数据写入到另外一个hbase表中
 */
public class TravelExample {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "hadoop");
        config.set("fs.defaultFS", "hdfs://192.168.1.103:9000");
        Job job = Job.getInstance(config, "travel");
        job.setJarByClass(TravelExample.class);
        // 注意,最后一个参数 addDependencyJars 在本地开发环境设置为false,在集群环境下设置为true.不能用简化的方法
        TableMapReduceUtil.initTableMapperJob("travel", new Scan(), TravelMapper.class, Text.class, TravelDataModel.class, job, false);
        // 同上
        TableMapReduceUtil.initTableReducerJob("online_product", TravelReducer.class, job, null, null, null, null, false);
        job.waitForCompletion(true);

    }


    private static Map<String, String> getContentMap(String content) {
        Map<String, String> map = new HashMap<>();
        int i = 0;
        String key = "";
        StringTokenizer tokenizer = new StringTokenizer(content, "({}\":,)");
        while (tokenizer.hasMoreTokens()) {
            if (++i % 2 == 0) {
                map.put(key, tokenizer.nextToken());
                // return tokenizer.nextToken();
            } else {
                // map.put()
                key = tokenizer.nextToken();
            }
        }
        return map;
    }

    /**
     * 从hbase输入数据
     */
    static class TravelMapper
            extends TableMapper<Text, TravelDataModel> {


        /**
         * @param key     rowKey
         * @param value   获取到的 rowKey 对应的一行数据
         * @param context map结果输出对象
         * @throws IOException          异常
         * @throws InterruptedException 异常
         */
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            //{"p_id":"100019","p_name":"<张家界-凤凰-天门山-玻璃栈道5日游>","price":"2141"}
            String content = Bytes.toString(value.getValue(Bytes.toBytes("f"), Bytes.toBytes("content")));
            if (content == null) {
                System.err.println("未获取到数据");
                return;
            }

            Text outputKey = new Text();
            TravelDataModel outputValue = new TravelDataModel();
            Map<String, String> contentMap = getContentMap(content);
            if (contentMap.containsKey("p_id") && contentMap.containsKey("p_name") && contentMap.containsKey("price")) {
                outputKey.set(contentMap.get("p_id"));
                outputValue.setId(contentMap.get("p_id"));
                outputValue.setName(contentMap.get("p_name"));
                outputValue.setPrice(contentMap.get("price"));
                // System.out.println("outputKey=" + Bytes.toString(outputKey.getBytes()));
                // System.out.println("outputValue=" + outputValue);

            } else {
                System.err.println("格式错误");
                return;
            }
            context.write(outputKey, outputValue);
        }
    }

    static class TravelReducer extends TableReducer<Text, TravelDataModel, ImmutableBytesWritable> {
        @Override
        protected void reduce(Text key, Iterable<TravelDataModel> values, Context context) throws IOException, InterruptedException {
            for (TravelDataModel value : values) {
                if (value != null) {
                    ImmutableBytesWritable outputKey = new ImmutableBytesWritable(key.getBytes());
                    Put put = new Put(key.getBytes());
                    // System.out.println(Bytes.toString(key.getBytes()));
                    // System.out.println(value);
                    put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("id"), Bytes.toBytes(value.getId()));
                    put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes(value.getName()));
                    put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("price"), Bytes.toBytes(value.getPrice()));
                    context.write(outputKey, put);
                }
            }
        }
    }
}


/**
 * 自定义数据格式
 *
 */
class TravelDataModel implements WritableComparable<TravelDataModel> {
    private String id;
    private String name;
    private String price;

    String getId() {
        return id;
    }

    void setId(String id) {
        this.id = id;
    }

    String getName() {
        return name;
    }

    void setName(String name) {
        this.name = name;
    }

    String getPrice() {
        return price;
    }

    void setPrice(String price) {
        this.price = price;
    }


    TravelDataModel() {
    }

    public TravelDataModel(String id, String name, String price) {
        this.id = id;
        this.name = name;
        this.price = price;
    }

    @Override
    public int compareTo(@NotNull TravelDataModel o) {
        return this.id.compareTo(o.id) + this.name.compareTo(o.name) + this.price.compareTo(o.price);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(this.id);
        out.writeUTF(this.name);
        out.writeUTF(this.price);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        id = in.readUTF();
        name = in.readUTF();
        price = in.readUTF();

    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof TravelDataModel)) return false;
        TravelDataModel that = (TravelDataModel) o;
        return Objects.equals(getId(), that.getId()) &&
                Objects.equals(getName(), that.getName()) &&
                Objects.equals(getPrice(), that.getPrice());
    }

    @Override
    public int hashCode() {
        return Objects.hash(getId(), getName(), getPrice());
    }

    @Override
    public String toString() {
        return "TravelDataModel{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", price='" + price + '\'' +
                '}';
    }
}
